package org.jmqtt.broker.client;

import io.netty.channel.Channel;
import org.apache.commons.lang3.StringUtils;
import org.jmqtt.broker.common.log.JmqttLogger;
import org.jmqtt.broker.common.log.LogUtil;
import org.jmqtt.broker.common.model.Message;
import org.jmqtt.broker.common.model.Subscription;
import org.jmqtt.broker.processor.dispatcher.InnerMessageDispatcher;
import org.jmqtt.broker.remoting.netty.ChannelEventListener;
import org.jmqtt.broker.remoting.session.ClientSession;
import org.jmqtt.broker.remoting.session.ConnectManager;
import org.jmqtt.broker.remoting.util.NettyUtil;
import org.jmqtt.broker.store.MessageStore;
import org.jmqtt.broker.store.SessionState;
import org.jmqtt.broker.store.SessionStore;
import org.jmqtt.broker.subscribe.SubscriptionMatcher;
import org.slf4j.Logger;

import java.util.Set;

public class ClientLifeCycleHookService implements ChannelEventListener {

    private static final Logger log = JmqttLogger.clientTraceLog;
    private MessageStore messageStore;
    private SessionStore sessionStore;
    private SubscriptionMatcher subscriptionMatcher;
    private InnerMessageDispatcher innerMessageDispatcher;

    public ClientLifeCycleHookService(SessionStore sessionStore, MessageStore messageStore,
                                      SubscriptionMatcher subscriptionMatcher, InnerMessageDispatcher innerMessageDispatcher) {
        this.sessionStore = sessionStore;
        this.messageStore = messageStore;
        this.subscriptionMatcher = subscriptionMatcher;
        this.innerMessageDispatcher = innerMessageDispatcher;
    }

    @Override
    public void onChannelConnect(String remoteAddr, Channel channel) {
    }

    @Override
    public void onChannelClose(String remoteAddr, Channel channel) {
        String clientId = NettyUtil.getClientId(channel);
        if (StringUtils.isNotEmpty(clientId)) {
            Message willMessage = messageStore.getWillMessage(clientId);
            if (willMessage != null) {
                innerMessageDispatcher.appendMessage(willMessage);
            }
        }
    }

    @Override
    public void onChannelIdle(String remoteAddr, Channel channel) {
        String clientId = NettyUtil.getClientId(channel);
        if (StringUtils.isNotEmpty(clientId)) {
            // 移除用户
            ConnectManager connectManager = ConnectManager.getInstance();
            ClientSession client = connectManager.getClient(clientId);
            if (null != client) {
                LogUtil.info(log, "[ClientLifeCycleHook] -> 心跳获取超时 『{}』 将被清理....", clientId);
                clearSession(client);
                connectManager.removeClient(clientId);
            }
        }
    }

    @Override
    public void onChannelException(String remoteAddr, Channel channel) {
        String clientId = NettyUtil.getClientId(channel);
        ConnectManager.getInstance().removeClient(clientId);
        LogUtil.warn(log, "[ClientLifeCycleHook] -> {} channelException,close channel and remove ConnectCache!", clientId);
    }

    private void clearSession(ClientSession clientSession) {
        if (clientSession.isCleanStart()) {
            Set<Subscription> subscriptions = sessionStore.getSubscriptions(clientSession.getClientId());
            for (Subscription subscription : subscriptions) {
                subscriptionMatcher.unSubscribe(subscription.getTopic(), clientSession.getClientId());
            }
            sessionStore.clearSession(clientSession.getClientId(), false);
        } else {
            SessionState sessionState = new SessionState(SessionState.StateEnum.OFFLINE, System.currentTimeMillis());
            sessionStore.storeSession(clientSession.getClientId(), sessionState);
        }
    }

}
